package com.bmslz.monitorpre.job;

import com.bmslz.monitorpre.grpc.MonitorGrpcClient;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.support.BasicAuthorizationInterceptor;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;
import org.springframework.web.util.UriUtils;

import java.net.URI;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Component
public class PushDataJob {
    @Autowired
    private MonitorGrpcClient monitorGrpcClient;
    @Value("${client.project.rabbitMQ.rabbitMQManagementUri}")
    private String rabbitMQManagementUri;

    @Value("${client.project.rabbitMQ.queues}")
    private String queues;
    @Value("${client.project.rabbitMQ.vhost:/}")
    private String vhost;
    @Autowired
    private RestTemplate restTemplate;
    private final ObjectMapper objectMapper;

    public PushDataJob() {
        this.objectMapper = new ObjectMapper();
    }

    /**
     * 测点数据推送demo
     * map中是监控模型的数据
     * 每一个key对应模型中的一个测点key,value为当前测点数据
     * Map<String, String> data = new HashMap<>();
     * data.put("cpu.use", "1");
     * data.put("mem.use", "100");
     * monitorGrpcClient.sendMetrics(data);
     * return "Metrics sent";
     */
    @Scheduled(fixedRateString = "#{${client.project.rate} * 1000}")
    public void executeTask() {
        //下面写你测点数据的获取逻辑:请求远程接口\获取本机信息\获取其他组件暴露的信息等方式都可以
        Map<String, String> map = getData();
        log.warn("上传数据-{}", map);
        monitorGrpcClient.sendMetrics(map);
    }

    private Map<String, String> getData() {
        Map<String, String> map = new HashMap<>();
        String[] queuesArr = queues.split(",");
        for (String queue : queuesArr) {
            Map<String, String> queueMetrics = getQueueMetrics(queue);
            map.put(queue + "." + "messages", queueMetrics.get("messages"));//消息总数
            map.put(queue + "." + "messages_ready", queueMetrics.get("messages_ready"));//消息堆积数
            map.put(queue + "." + "messages_unacknowledged", queueMetrics.get("messages_unacknowledged"));//未ack数
            map.put(queue + "." + "consumers", queueMetrics.get("consumers"));//消费者数量
        }
        return map;
    }

    /**
     * 获取指定队列的监控指标
     */
    public Map<String, String> getQueueMetrics(String queueName) {
        Map<String, String> metrics = new HashMap<>();

        try {

            String encodedVhost = URLEncoder.encode(vhost);
            String url = "";
            url = rabbitMQManagementUri + "/api/queues/" + encodedVhost + "/" + queueName;
            URI uri = URI.create(url);
            log.warn("开始抓取-{}", url);
            ResponseEntity<String> response = restTemplate.getForEntity(uri, String.class);

            if (response.getStatusCode() == HttpStatus.OK) {
                JsonNode queueInfo = objectMapper.readTree(response.getBody());

                // 提取各项指标
                metrics.put("queue_name", queueName);
                metrics.put("messages", queueInfo.path("messages").asText()); // 队列中总消息数（待处理 + 未确认）
                metrics.put("messages_ready", queueInfo.path("messages_ready").asText()); // 待处理消息数（可被消费的消息）
                metrics.put("messages_unacknowledged", queueInfo.path("messages_unacknowledged").asText()); // 未确认消息数
                metrics.put("consumers", queueInfo.path("consumers").asText()); // 消费者数量
                // 其他有用指标
                metrics.put("state", queueInfo.path("state").asText()); // 队列状态
                metrics.put("memory", queueInfo.path("memory").asText()); // 内存使用量（字节）
                metrics.put("messages_ram", queueInfo.path("messages_ram").asText()); // RAM 中的消息数
                metrics.put("messages_persistent", queueInfo.path("messages_persistent").asText()); // 持久化消息数
            }
        } catch (Exception e) {
            log.error("队列抓取失败-{}", e.getMessage());
        }

        return metrics;
    }
}

